package com.controller.consumer.hbase;

import com.bean.consume.HBaseBean;
import com.dms.conf.DmsConfig;
import com.dms.consumer.DmsConsumer;
import com.service.consume.HBaseService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;

import java.util.Arrays;

@Controller
@RequestMapping("/hb")
public class HBaseController {

    @Autowired
    private HBaseService hBaseService;

    @RequestMapping("/consume")
    @ResponseBody
    public String consumerHBase() throws Exception {
        String path = DmsConfig.CONFIG_CONSUMER_FILE_NAME;
        String groupId = "group_HBase";
        DmsConsumer consumer = new DmsConsumer(groupId, path);
        consumer.consume(Arrays.asList("topic-Server", "topic-Log"));
        StringBuilder sb = new StringBuilder();
        ConsumerRecords<Object, Object> records = consumer.poll(1000);
        System.out.println("the numbers of topic:" + records.count());
        sb.append("the numbers of topic:").append(records.count());
        for (ConsumerRecord<Object, Object> record : records) {
            hBaseService.insertHBaseBean(new HBaseBean(record));
            System.out.println("consume HBase:" + record.toString());
            sb.append(record);
        }
        return sb.toString();
    }

}
